package com.lagou.es.service.impl;

import com.lagou.es.service.PositionService;
import com.lagou.es.utils.DBHelper;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.bulk.BackoffPolicy;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.MatchQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;

@Service
@Slf4j
public class PositionServiceImpl implements PositionService {
    
    @Autowired
    private RestHighLevelClient client;
    private static final String POSITION_INDEX = "position";

    @Override
    public List<Map<String, Object>> searchPos(String keyword, String field, int pageNo, int pageSize) {
        if (pageNo <= 0) pageNo = 1;
        // 条件搜索
        SearchRequest searchRequest = new SearchRequest(POSITION_INDEX);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        // 分页 index = (当前页 - 1) * 页容量
        searchSourceBuilder.from((pageNo - 1) * pageSize);
        searchSourceBuilder.size(pageSize);
        // 精准匹配
        MatchQueryBuilder builder = QueryBuilders.matchQuery(field, keyword);
        searchSourceBuilder.query(builder);
        searchSourceBuilder.timeout(new TimeValue(60,TimeUnit.SECONDS));
        searchRequest.source(searchSourceBuilder);
        List<Map<String,Object>> result = new ArrayList<>();
        try {
            // 执行搜索
            SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
            SearchHit[] hits = searchResponse.getHits().getHits();
            for (SearchHit hit : hits) {
                result.add(hit.getSourceAsMap());
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

        return result;
    }

    @Override
    public void importAll() {
        writeMysqlData2ES(POSITION_INDEX);
    }

    /**
     * 将mysql中的数据批量写入ES
     */
    private void writeMysqlData2ES(String tableName){
        BulkProcessor bulkProcessor = getBulkProcessor();
        Connection conn = null;
        PreparedStatement ps = null;
        ResultSet rs = null;
        try {
            conn = DBHelper.getConnection();
            log.info("Start handle data {}",tableName);
            String sql = "select * from " + tableName;
            ps = conn.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY,ResultSet.CONCUR_READ_ONLY);
            ps.setFetchSize(20);
            rs = ps.executeQuery();
            ResultSetMetaData columnData = rs.getMetaData();
            List<Map<String, String>> dataList = new ArrayList<>();
            // bulkProcessor 添加的数据支持的方式并不多，查看其api发现其支持map键值对的方式。
            // 故在此将查出来的数据转换成hashMap方式。
            Map<String,String> map = null;
            int count = 0;
            String c;
            String v;
            while (rs.next()){
                count++;
                map = new HashMap<>(128);
                for (int i = 1; i <= columnData.getColumnCount(); i++){
                    c = columnData.getColumnName(i);
                    v = rs.getString(c);
                    map.put(c,v);
                }
                dataList.add(map);
                // 每1万条写一次
                if (count%10000 == 0){
                    log.info("mysql handle data number = [{}]",count);
                    // 将数据添加到bulkProcessor中
                    for (Map<String,String> data : dataList){
                        System.out.println(data);
                        bulkProcessor.add(new IndexRequest(POSITION_INDEX).source(data));
                    }
                    // 每提交一次便将map与list清空
                    map.clear();
                    dataList.clear();
                }
            }
            
            // 处理未提交的数据
            for (Map<String,String> data : dataList){
                System.out.println(data);
                bulkProcessor.add(new IndexRequest(POSITION_INDEX).source(data));
            }
            log.info("-----Finally insert number total = [{}]",count);
            // 将数据刷新到es，注意这一步执行后不会立即生效，取决于bulkProcessor设置到刷新时间
            bulkProcessor.flush();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                rs.close();
                ps.close();
                conn.close();
                boolean terminateFlag = bulkProcessor.awaitClose(150L, TimeUnit.SECONDS);
                log.info("terminateFlag = [{}]",terminateFlag);
            } catch (Exception e){
                e.printStackTrace();
            }
        }
    }
    
    private BulkProcessor getBulkProcessor(){
        BulkProcessor bulkProcessor = null;
        try {
            BulkProcessor.Listener listener = new BulkProcessor.Listener() {
                @Override
                public void beforeBulk(long executionId, BulkRequest bulkRequest) {
                    log.info("Try to insert data number = [{}]",bulkRequest.numberOfActions());
                }
    
                @Override
                public void afterBulk(long executionId, BulkRequest bulkRequest, BulkResponse bulkResponse) {
                    log.info("Success insert data number : [{}] , id : [{}]",bulkRequest.numberOfActions(),executionId);
                }
    
                @Override
                public void afterBulk(long executionId, BulkRequest bulkRequest, Throwable throwable) {
                    log.error("Bulk is fail : [{}] , executionId : [{}]",throwable,executionId);
                }
            };
            BiConsumer<BulkRequest, ActionListener<BulkResponse>> bulkConsumer = (request, bulkListener) -> client
                    .bulkAsync(request, RequestOptions.DEFAULT,bulkListener);
            BulkProcessor.Builder builder = BulkProcessor.builder(bulkConsumer, listener);
            builder.setBulkActions(5000);
            builder.setBulkSize(new ByteSizeValue(100L, ByteSizeUnit.MB));
            builder.setConcurrentRequests(10);
            builder.setFlushInterval(TimeValue.timeValueSeconds(100L));
            builder.setBackoffPolicy(BackoffPolicy.constantBackoff(TimeValue.timeValueSeconds(1L),3));
            //注意点：让参数设置生效
            bulkProcessor = builder.build();
        } catch (Exception e) {
            e.printStackTrace();
            try {
                bulkProcessor.awaitClose(100L, TimeUnit.SECONDS);
            } catch (InterruptedException interruptedException) {
                e.printStackTrace();
                log.error(interruptedException.getMessage());
            }
        }
        return bulkProcessor;
    }
}
